\lstset{language=C++}
\begin{lstlisting}[caption=cast.cc]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.

#include "cast.h"

#include <math.h>

#include <iostream>
#include <sstream>
#include <string>

#include "pigasus-types.h"

using namespace std;

int CastToInt(const int &v) {
  return v;
}

int CastToInt(const double &v) {
  return round(v);
}

int CastToInt(const string &v) {
  int buffer;
  sscanf(v.c_str(), "%d", &buffer);
  return buffer;
}

double CastToFloat(const int &v) {
  return (double)v;
}

double CastToFloat(const double &v) {
  return v;
}

double CastToFloat(const string &v) {
  double buffer;
  sscanf(v.c_str(), "%lf", &buffer);
  return buffer;
}

string CastToString(const int &v) {
  stringstream out;
  out << v;
  return out.str();
}

string CastToString(const double &v) {
  stringstream out;
  out << v;
  return out.str();
}

string CastToString(const string &v) {
  stringstream out;
  out << v;
  return out.str();
}

File CastToFile(const string &v) {
  return File(v);
}

int CastToInt(void *v) {
  return *((int*)v);
}

double CastToFloat(void *v) {
  return *((double*)v);
}

string CastToString(void *v) {
  return *((string*)v);
}

List CastToList(void *v) {
  return ((List*)v)->CloneValue();
}

Map CastToMap(void *v) {
  return ((Map*)v)->CloneValue();
}

File CastToFile(void *v) {
  return ((File*)v)->CloneValue();
}

List CastToList(const List &v) {
  return v;
}

Map CastToMap(const Map &v) {
  return v;
}

File CastToFile(const File &v) {
  return v;
}

\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=cast.h]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.
//
// Functions to facilitate casting between types. 

#ifndef CAST_H_ 
#define CAST_H_

#include <string>

#include "pigasus-types.h"

using namespace std;

// Cast a C++ type to an int.
int CastToInt(const int &v);
int CastToInt(const double &v);
int CastToInt(const string &v);

// Cast a C++ type to a double (Pigasus float).
double CastToFloat(const int &v);
double CastToFloat(const double &v);
double CastToFloat(const string &v);

// Cast a C++ type to a string.
string CastToString(const int &v);
string CastToString(const double &v);
string CastToString(const string &v);

// Cast a string to File; the only way to create a File in Pigasus.
File CastToFile(const string &v);

// Use the following casting functions when casting the result of List.Get() or
// Map.Get().
int CastToInt(void *v);
double CastToFloat(void *v);
string CastToString(void *v);
List CastToList(void *v);
Map CastToMap(void *v);
File CastToFile(void *v);

// Allow Lists, Files and Maps to cast to themselves.
List CastToList(const List &v);
Map CastToMap(const Map &v);
File CastToFile(const File &v);

#endif  // CAST_H_
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=checkdone.sh]
#!/usr/bin/env bash
# Author: Jordan Rupprecht

EXEC="ssh $1@$2 'ls pig-$3/done' 2> /dev/null"
if [ -n "`eval $EXEC`" ]; then
echo 1
else
echo 0
fi
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=common.cc]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.

#include <fstream>
#include <sstream>
#include <string>

using namespace std;

string IntToString(int i) {
  stringstream ss;
  ss << i;
  return ss.str();
}

bool ReadStringFromFile(string filename, string &contents) {
  ifstream file_stream(filename.c_str());
  stringstream ss;
  ss << file_stream.rdbuf();

  if (!file_stream && !file_stream.eof()) {
    // Error reading file!
    return false;
  }

  contents = ss.str();
  return true;
}

bool WriteStringToFile(string filename, string &contents) {
  ofstream file_stream;
  file_stream.open(filename.c_str());
  file_stream << contents;
  file_stream.close();
  return true;
}
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=common.h]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.
//
// Defines a number of common subroutines and macros that all pigasus-related
// code should include.
//
// NOTE: To enable debug messages, compile using -DENABLE_DEBUG=1 flag.

#ifndef COMMON_H_ 
#define COMMON_H_

#include <stdio.h>

#include <sstream>
#include <string>

using namespace std;

// Compile with -DENABLE_DEBUG to get debug code
#ifdef ENABLE_DEBUG
  #define debug(args...) fprintf (stderr, args)
#else
  #define debug(args...)
#endif

// Converts int to C++ string.
string IntToString(int i);

// Read an entire file into a string. Normally a bad idea, so we'll implement
// something smarter later. Returns false if there's an error, true otherwise.
bool ReadStringFromFile(string filename, string &contents);

// Writes a string to a file.
bool WriteStringToFile(string filename, string &contents);

#endif // COMMON_H_
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=formatstring.cc]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.

#include "formatstring.h"

#include <stdlib.h>
#include <string.h>

#include <iostream>
#include <string>

#include "pigasus-types.h"

using namespace std;

#define BUFFER_SIZE 1000

// Given a format string fmt that is known to only contain one format specifier
// as well as the argument that is to be put into the string and the associated
// type, PrintChunk returns the formatted string (i.e. result of sprintf()).
// However, if fmt contained %%, then arg is actually not used, so we set
// arg_used to 0 and return. Otherwise, it is set to 1.
string PrintChunk(string fmt, void *arg, string type, int *arg_used) {
  char buf[BUFFER_SIZE];
  
  if (type == kIntType) {
    sprintf(buf, fmt.c_str(), *((int*)arg));
  } else if (type == kFloatType) {
    sprintf(buf, fmt.c_str(), *((double*)arg));
  } else if (type == kStringType) {
    sprintf(buf, fmt.c_str(), ((string*)arg)->c_str());
  } else {
    sprintf(buf, fmt.c_str()); // No more args.
  }

  string res(buf);
  if (res.find("%") != string::npos) {
    // That means fmt contained "%%" and arg was not used.
    *arg_used = 0;
  } else {
    *arg_used = 1;
  }
  return res;
}

#undef BUFFER_SIZE

// The FormatString function takes a printf() format string and a Pigasus List
// which should only contain ints, Floats, and string's, and returns the
// formatted string. This function works as follows. The general idea is to
// break the format string up into sections so that teach section contains only
// one format specifier (i.e. %something) and apply the next element of the List
// to that format string, and concatenate that to a result string, which is
// returned at the end. 
string FormatString(string fmt, const List &arg) {
  size_t found;
  string res;
  int prev_s = 0;
  int prev_m = 0;
  int used;
  int i = 0;  // Counter for args.

  List *args = (List*)(&arg);
  
  while (true) {
    // Look for the next occurrence of %.
    found = fmt.find("%", prev_s);
    if (found == string::npos) {
      // No other %'s found, so the rest of the format string can be taken as
      // is.
      res += fmt;
      break;
    } else {
      prev_m = int(found);
      
      // One occurrence of % has been found, now find the one after that.
      found = fmt.find("%", prev_m + 1);
      if (int(found) == prev_m + 1) {
        // We found a %%, so the second % doesn't count.
        found = fmt.find("%", prev_m + 2);
      }

      if (found == string::npos) {        
        // Only one % left in the remaining string. So take that chunk.
        res += PrintChunk(fmt.substr(prev_s, fmt.size() - prev_s + 1),
                          args->Get(i), args->GetType(i), &used);
        if (used) {
          i++;                 
        }
        break;
      } else {
        // Take the chunk containing the % at prev_m, which is located between
        // indices prev_s and int(found) and use that.
        res += PrintChunk(fmt.substr(prev_s, int(found) - prev_s),
                          args->Get(i), args->GetType(i), &used);
        if (used) {
          i++;
        }
        prev_s = int(found);
      }
    }
  }

  return res;
}

\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=formatstring.h]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.
//
// Defines the function needed by the <string> % <list> operator (i.e.
// equivalent to the Python % operator).
//
// User should call FormatString(), never PrintChunk().

#ifndef FORMATSTRING_H_
#define FORMATSTRING_H_

#include <string>

#include "pigasus-types.h"

using namespace std;

// Applies sprintf() to fmt using arg, where fmt is a format string that is
// known to contain exactly one format specifier (it may be %%). Also send in a
// flag arg_used, which will be set to 1 if arg is actually used (i.e. %% is not
// in fmt) or 0 otherwise.
string PrintChunk(string fmt, void *arg, string type, int *arg_used);

// Applies <fmt> % <args>, or similar to printf(fmt, args). 
string FormatString(string fmt, const List &arg);

#endif  // FORMATSTRING_H_
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=getlength.h]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.
//
// Defines the GetLength (List or Map) function

#ifndef GETLENGTH_H_
#define GETLENGTH_H_

#include "pigasus-types.h"

static inline int GetLength(const List &l) {
  return ((List*)(&l))->GetSize();
}

static inline int GetLength(const Map &m) {
  return ((Map*)(&m))->GetSize();
}

#endif // GETLENGTH_H_
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=getthread.cc]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.

#include "getthread.h"

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>

#include <fstream>
#include <sstream>
#include <string>

#include "common.h"
#include "pigasus-types.h"
#include "threadpool.h"

using namespace std;

extern ThreadPool *get_thread_pool;
extern pthread_mutex_t *mutex_get_thread_pool;

GetThread::GetThread(int id_to_get_, Map &server_, Map *output_) {
  debug("GetThread constructor\n");
  id_to_get = id_to_get_;
  server = &server_;
  output = output_;
}

GetThread::~GetThread() { }

void GetThread::Run() {
  // Get the list of files
  // TODO: retrieve the list of files. Right now well just get pig-<id>/*
  // We could maybe have a list of files that have changed, get that, and read
  // it in; then we would grab each of those files.
  FILE *p;
  char ignore[20];
  char *err;
  string buffer;  //TODO: dynamic size
  string user = *(string*)server->Get("user");
  string host = *(string*)server->Get("host");
  debug("GetThread starting\n");
  
  // Prepare the directory for the files
  buffer = "mkdir -p out-" + IntToString(id_to_get);
  p = popen(buffer.c_str(), "r");
  while (!feof(p)) err = fgets(ignore, sizeof(ignore), p);
  pclose(p);
  
  // Get the files
  buffer = "scp " + user + "@" + host + ":pig-" + IntToString(id_to_get) +
           "/changed.tar.gz ./out-" + IntToString(id_to_get);
  debug("Buffer for copying tar over is %s\n", buffer.c_str());
  
  p = popen(buffer.c_str(), "r");
  while (!feof(p)) err = fgets(ignore, sizeof(ignore), p);
  pclose(p);

  // Delete the temporary pig-directory in the remote computer.
  buffer = "ssh -nT " + user + "@" + host + " \"rm -rf pig-" +
           IntToString(id_to_get) + "\"";
  debug("Buffer for deleting remote directory is %s\n", buffer.c_str());

  p = popen(buffer.c_str(), "r");
  while (!feof(p)) err = fgets(ignore, sizeof(ignore), p);
  pclose(p);
            
  // Untar and get list of changed files
  buffer = "tar -zxvf out-" + IntToString(id_to_get) + "/changed.tar.gz "
  "-C out-" + IntToString(id_to_get) + " > out-" + IntToString(id_to_get) + "/changed";
  p = popen(buffer.c_str(), "r");
  while (!feof(p)) err = fgets(ignore, sizeof(ignore), p);
  pclose(p);
            
  buffer = "./out-" + IntToString(id_to_get);
  debug("Root is %s, output is at %p\n", buffer.c_str(), output);
  debug("Size is %d\n", output->GetSize());
  output->Set("root", buffer);
    
  string stdout_file;
  ReadStringFromFile(buffer + "/stdout", stdout_file);
  output->Set("stdout", stdout_file);
  debug("GetThread %d: Stdout = %s\n", id, stdout_file.c_str());
  
  string stderr_file;
  ReadStringFromFile(buffer + "/stderr", stderr_file);
  output->Set("stderr", stderr_file);
  debug("GetThread %d: stderr = %s\n", id, stderr_file.c_str());
  
  List files;
  string changed_path = buffer + "/changed";
  ifstream ifs(changed_path.c_str());
  while (ifs.good()) {
    string s;    
    getline(ifs, s);

    if ((s == "stderr") || (s == "stdout") || (s.empty())) {
      continue;
    }

    files.Append(s);
  }
  output->Set("files", files);
  
  //Set status to done
  pthread_mutex_lock(mutex_get_thread_pool);
  get_thread_pool->SetStatus(id, kThreadStatusMarkAsDone);
  pthread_mutex_unlock(mutex_get_thread_pool);
}
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=getthread.h]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.
//
// Defines the GetThread class, which is used for getting files from
// the remote server

#ifndef GETTHREAD_H_
#define GETTHREAD_H_

#include "thread.h"
#include "pigasus-types.h"

class GetThread : public Thread {
  int id_to_get;
  Map *server;
  Map *output;
  
 public:
  GetThread(int id_to_get, Map &server, Map *output);
  ~GetThread();
  void Run();
};

#endif // GETTHREAD_H_
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=pigasus-grammar.ypp]
/*==============================================================================
pigasus-grammar.ypp
Bison Parser for the Pigasus Language
By: Philip Tjimos and Sharath Gururaj

==============================================================================*/
%{
int yylex(void);
int yyerror(const char *s);
#define YYSTYPE string
#include <stdio.h>
#include <string>
#include <iostream>
#include <fstream>
#include <map>
#include <string.h>
static int yyline = 1;
using namespace std;
map<string, string> funcMap;
map<string, string> jobMap;
string headers = 
    "#include \"cast.h\"\n"
    "#include \"jobthread.h\"\n"
    "#include \"pigasus-types.h\"\n"
    "#include \"serialization.h\"\n"
    "#include \"threadpool.h\"\n"
    "#include \"formatstring.h\"\n"
    "#include \"loadcsv.h\"\n"
    "#include \"servers.h\"\n"
    "#include \"getthread.h\"\n"
    "#include \"getlength.h\"\n"
    "#include \"wait.h\"\n"
    "#include \"sendthread.h\"\n"
    "#include \"thread.h\"\n"
    "#include \"push.h\"\n"
    "#include \"system.h\"\n"
    "#include \"common.h\"\n"
    "#include <string>\n"
    "#include <iostream>\n"
    "using namespace std;\n"
    "extern ThreadPool *job_thread_pool;\n"
    "extern pthread_mutex_t *mutex_job_thread_pool;\n"
    "extern ThreadPool *send_thread_pool;\n"
    "extern pthread_mutex_t *mutex_send_thread_pool;\n"
    "extern ThreadPool *get_thread_pool;\n"
    "extern pthread_mutex_t *mutex_get_thread_pool;\n"
    "extern Servers *global_servers;\n\n\n"
    ;
string headers2 =
    "#include \"cast.h\"\n"
    "#include \"pigasus-types.h\"\n"
    "#include \"serialization.h\"\n"
    "#include \"formatstring.h\"\n"
    "#include \"loadcsv.h\"\n"
    "#include \"getlength.h\"\n"
    "#include \"system.h\"\n"
    "#include \"common.h\"\n"
    "#include <iostream>\n"
    "#include <string>\n"
    "using namespace std;\n\n\n" 
    ;
string waitFooter = 
    "while (true) {\n"
    "bool a = job_thread_pool->HasJobsRunning();\n"
    "bool b = send_thread_pool->HasJobsRunning();\n"
    "bool c = get_thread_pool->HasJobsRunning();\n"
    "if (!(a || b || c)) break;\n"
    "sleep(1);\n"
    "}\n"
    ;

%}
%token IDENTIFIER CONSTANT STRING_LITERAL
%token AND_OP OR_OP LE_OP GE_OP EQ_OP NE_OP

%token BREAK CONNECT CONTINUE DO ELSE FILES FLOAT FOR IF INT JOB
%token LENGTH LIST LOADCSV MAP MAPS_TO NULLS PRINT PUSH RETURN 
%token SIZEOF STRING VOID WAIT WAITALL WHILE MOD MAIN

%left '+'


%%
compilation_unit    : translation_unit  {

  map<string, string>::const_iterator iter;
  ofstream myfile;
  string mainFunctionPartA = "int main(int argc, char** argv) {"
    "string buffer;\n"
    "ReadStringFromFile(string(argv[1]), buffer);\n"  
    "List *inputs = UnserializeList(buffer);\n";
  
  string mainFunctionPartB =  "(*inputs);\n"  
    "delete inputs;\n"
    "return 0;\n"
    "}\n";
    for (iter=jobMap.begin(); iter != jobMap.end(); ++iter) {
        myfile.open((iter->first+".cc").c_str());
    myfile<<headers2;
    myfile << (iter->second) ;
    
    map<string, string>::const_iterator iter2;
    for(iter2=funcMap.begin(); iter2 != funcMap.end(); ++iter2)
    {
      //do not copy pigmain to job files
      if(iter2->first != "PigMain")
        myfile << (iter2->second);
    }
    myfile<<mainFunctionPartA+iter->first+mainFunctionPartB;
    myfile.close();
    }
  //output the locals file
  map<string, string>::const_iterator iter2;
  myfile.open(("pigMain.cc"));
  myfile<<headers;
  for (iter2=funcMap.begin(); iter2 != funcMap.end(); ++iter2) {
    myfile<<iter2->second;
  }
  myfile.close();

}
                    ;

translation_unit    : translation_unit function_definition 
                      { string s = $1 + $2 + "\n"; printf("%s", s.c_str());}
                    | translation_unit job_definition      
                      {string s = $1 + $2 + "\n"; printf("%s", s.c_str());}
                    |
                    ;
                                      
job_definition      : JOB IDENTIFIER '(' LIST IDENTIFIER ')' compound_statement
                        { $$ = "void "+ $2 + "( " + "List " + $5 + " ) " + $7; 
                        jobMap[$2] = $$;}
                    ;

/*the declaration part*/
declaration         : primary_declaration
                    | function_definition
                    ;

primary_declaration : type_specifier identifier_list { $$ = $1 + " " + $2; }
                    ;

identifier_list     : init_statement        
                    | identifier_list ',' init_statement  { $$ = $1 + " " + ',' + " " + $3; }
                    ;
                    
init_statement      : IDENTIFIER
                    | IDENTIFIER '=' assignment_expression   { $$ = $1 + " " + '=' + " " + $3; }
                    | IDENTIFIER '=' push_statement      { $$ = $1 + " = " + $3; }
                    ;
                    
list_initializer    : '[' list_elements ']'   { $$ = $2;  }
                    ;
                    
map_initializer     : '[' map_element_list ']'  { $$ = $2; }
                    ;

list_elements       : element                     
                      { $$ = "NewList( " + $1 + " )"; }
                    | element ',' list_elements   
                      { $$ = "NewList( " + $1 + " ) + " + $3; }
                    ;
                    
map_element_list    : map_element                 
                    | map_element ',' map_element_list
                      { $$ = $1 + " " + "+" + " " + $3; } 
                    ;
                    
map_element         : IDENTIFIER MAPS_TO assignment_expression     
                      { $$ = "Map( " + $1 + "," + $3 + " )"; }
                    | STRING_LITERAL MAPS_TO assignment_expression 
                      { $$ = "Map( " + $1 + "," + $3 + " )"; }
                    ; 
                    
element             : assignment_expression
                    ;
                    
function_definition     : type_specifier IDENTIFIER '(' formal_id_list ')' 
                            compound_statement
                            { $$ = $1 + " " + $2 + " " + '(' + " " + $4 + " " +
                              ')' + " " + $6; funcMap[$2]=$$; 
                              headers = headers + $1 + " " + $2 + " ( " + $4 + " ) ;\n";  
                              headers2 = headers2 + $1 + " " + $2 + " ( " + $4 + " ) ;\n";
                            }
                        | type_specifier IDENTIFIER '(' ')' compound_statement
                            { $$ = $1 + " " + $2 + " ( ) " + $5 ;funcMap[$2]=$$; 
                              headers = headers + $1 + " " + $2 + " ( ) ;\n";
                              headers2 = headers2 + $1 + " " + $2 + " ( ) ;\n";
                            }
                        | VOID MAIN '(' LIST IDENTIFIER ')' compound_statement_main
                            { $$ = "void PigMain ( List &" + $5 + " ) " +
                              $7; funcMap["PigMain"]=$$; }
                        | VOID MAIN '(' ')' compound_statement_main
                            { $$ = "void PigMain ( List &args ) " + $5;
                             funcMap["PigMain"]=$$; }
                        ;
                        
                      
formal_id_list      : type_specifier IDENTIFIER
                        { $$ = $1 + " " + $2; }
                    | formal_id_list ',' type_specifier IDENTIFIER
                        { $$ = $1 + " , " + $3 + " " + $4; }
                    ;

type_specifier      : VOID    {$$ = "void"; }
                    | INT     {$$ = "int"; }
                    | FLOAT   {$$ = "double"; }
                    | STRING  {$$ = "string"; }
                    | LIST    {$$ = "List"; }
                    | MAP     {$$ = "Map"; }
                    | FILES   {$$ = "File"; }
                    ;
                                        
/* end of our declarations*/

statement_list      : statement
                    | statement_list statement  { $$ = $1 + " " + $2; }
                    ;


statement           : matched_statement
                    | open_statement
                    | iteration_statement
                    ;
                                             
matched_statement         : compound_statement
                          | push_statement
                          | expression_statement
                          | jump_statement
                          | wait_statement
                          | PRINT expression ';'        
                              {$$ = "cout << ( "+$2+" ) ;\n"; } 
                          | IF '(' expression ')' matched_statement
                              ELSE matched_statement
                              { $$ = "if ( " + $3 + ") " + $5 + "else " + $7; }
                          ;
                                        
open_statement        : IF '(' expression ')' statement
                        { $$ = "if ( " + $3 + ") " + $5; }
                      | IF '(' expression ')' matched_statement 
                          ELSE open_statement
                        { $$ = "if ( " + $3 + ") " + $5 + "else " + $7; } 
                      ;                           
                    
      
                     
compound_statement  : '{' '}'  { $$ = "{ }\n"; }
                    | '{' statement_list '}'  { $$ = "{\n" + $2 + "}\n"; }
                    ;
                    
compound_statement_main : '{' statement_list '}'  
                        { $$ = "{\n" + $2 + waitFooter +"}\n";}


expression_statement  : ';'               { $$ = ";\n"; }
                      | expression ';'    { $$ = $1 + " " + ";\n"; }
                      | declaration ';'   { $$ = $1 + " " + ";\n"; }
                      ;
    

iteration_statement : WHILE '(' expression ')' statement
                        { $$ = "while ( " + $3 + " ) " + $5; }
                    | DO statement WHILE '(' expression ')' ';'
                        { $$ = "do " + $2 + " while ( " + $5 + " ) ;" }
                    | FOR '(' expression_statement  expression_statement ')'
                              statement
                        { $$ = "for ( " + $3 + " " + $4 + " ) " + $6; }
                    | FOR '(' expression_statement  expression_statement 
                        expression ')' statement
                        { $$ = "for ( " + $3 + " " + $4 + " " + $5 + " ) " + 
                          $7; }
                    ;



jump_statement    : CONTINUE ';' { $$ = "continue ;\n"; }
                  | BREAK ';'    { $$ = "break ;\n"; }
                  | RETURN ';'   { $$ = "return ;\n"; }
                  | RETURN expression ';' { $$ = "return " + $2 + " ;\n"; }
                  ;
               
push_statement    : PUSH assignment_expression ',' assignment_expression ','
                     assignment_expression ',' IDENTIFIER
                      { if ( strcmp($4.c_str(),"null") == 0 )
                          $4 = "List()";
                        if ( strcmp($6.c_str(),"null") == 0 )
                          $6 = "List()";
                        $$ = "Push ( \"" + $2 + "\" , " + $4+ " , " + $6 + 
                          " , &" + $8 + " );";           
                      }
                        
                  ;
                  
wait_statement    : WAIT IDENTIFIER ';' {$$ = "Wait ( " + $2 + " ) ;\n";}
                  | WAITALL ';'         {$$ = "WaitAll ( ) ;\n"; }
                  ;
                                    
primary_expression  : IDENTIFIER                    {$$ = $1;}
                    | CONSTANT                      {$$ = $1;}
                    | STRING_LITERAL                {$$ = $1;}
                    | NULLS                         {$$ = "null";}
                    | '(' expression ')'            {$$ = "( " + $2 + " )";}
                    | list_initializer
                    | map_initializer
                    ;
postfix_expression  : primary_expression                                      
                    | postfix_expression '[' expression ']'                   
                        {$$ = $1+ " " +"["+ " " +$3+ " " +"]";}
                    | postfix_expression '(' ')'                              
                        {$$ = $1 + " " + "()";}
                    | postfix_expression '(' argument_expression_list ')'     
                        {$$ = $1 + " " + "("+ " " +$3+ " " +")";}
                    ;
argument_expression_list  : assignment_expression
                          | argument_expression_list ',' assignment_expression    
                              {$$ = $1 + " " + "," + " " + $3;}
                          ; 
                          
unary_expression   : postfix_expression               {$$ = $1;}
                   | unary_operator cast_expression   {$$ = $1 + $2;}          
                   | LOADCSV cast_expression          {$$ = "LoadCSV("+$2+")";}    
                   | LENGTH cast_expression           {$$ = "GetLength( "+$2+" )";}   
                   | CONNECT cast_expression          
                   {$$ = "global_servers->SetServers("+$2+")";}   
                   | system_expression               
                   ;

system_expression  : '@' '{' expression '}'   { $$ = "System ( " + $3 + " )"; }
                   ;
                    
unary_operator     : '+'
                   | '-'
                   | '!'
                   ;
                    
cast_expression    : unary_expression
                   | '(' INT ')' cast_expression 
                        { $$ = "CastToInt( " + $4 + " ) "; }
                   | '(' FLOAT ')' cast_expression 
                        { $$ = "CastToFloat( " + $4 + " ) "; }
                   | '(' MAP ')' cast_expression 
                        { $$ = "CastToMap( " + $4 + " ) "; }
                   | '(' LIST ')' cast_expression 
                        { $$ = "CastToList( " + $4 + " ) "; }
                   | '(' FILES ')' cast_expression 
                        { $$ = "CastToFile( " + $4 + " ) "; }
                   | '(' STRING ')' cast_expression 
                        { $$ = "CastToString( " + $4 + " ) "; }
                   ;

multiplicative_expression : cast_expression
                          | multiplicative_expression '*' cast_expression 
                              {$$ = $1 + " " + "*" + " " + $3;}
                          | multiplicative_expression '/' cast_expression 
                              {$$ = $1 + " " + "/" + " " + $3;}
                          | multiplicative_expression '%' cast_expression 
                              {$$ = "FormatString("+ $1 + " , "  + $3 + ")";}
                          | multiplicative_expression MOD cast_expression 
                              {$$ = $1 + " % "  + $3 ;}
                          ;

additive_expression   : multiplicative_expression
                      | additive_expression '+' multiplicative_expression 
                        {$$ = $1 + " + " + $3;}
                      | additive_expression '-' multiplicative_expression 
                        {$$ = $1 + " - " + $3;}
                      ;

relational_expression : additive_expression
                      | relational_expression '<' additive_expression     
                        {$$ = $1 + " < " + $3;}
                      | relational_expression '>' additive_expression     
                        {$$ = $1 + " > " + $3;}
                      | relational_expression LE_OP additive_expression   
                        {$$ = $1 + " <= " + $3;}
                      | relational_expression GE_OP additive_expression   
                        {$$ = $1 + " >= " + $3;}
                      ;

equality_expression   : relational_expression
                      | equality_expression EQ_OP relational_expression   
                        {$$ = $1 + " == " + $3;}
                      | equality_expression NE_OP relational_expression   
                        {$$ = $1 + " != " + $3;}
                      ;

logical_and_expression  : equality_expression
                        | logical_and_expression AND_OP equality_expression 
                          {$$ = $1 + " " + "&&" + " " + $3;}
                        ;

logical_or_expression   : logical_and_expression
                        | logical_or_expression OR_OP logical_and_expression
                          {$$ = $1 + " " + "||" + " " + $3;}
                        ;
                        
assignment_expression   : unary_expression '=' assignment_expression  
                          {$$ = $1 + " = " + $3;}
                        | logical_or_expression       
                        ;
                        
                                               
expression              : assignment_expression   {$$ = $1; }
                        | expression ',' assignment_expression  
                          {$$ = $1 + " , " + $3; }
                        ;

%%
#include "lex.yy.cc"
int yyerror (const char* s)  /* Called by yyparse on error */
{
  fprintf(stderr, "Error at line %d: %s\n", yyline, s);
  return 1;
}

FlexLexer* lexer = new yyFlexLexer;
main()
{
  return yyparse();
}

int yylex()
{
  return lexer->yylex();
}
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=pigasus.lpp]
/*==============================================================================
pigasus.l
Lex Scanner for the Pigasus Language
By: Philip Tjimos and Sharath Gururaj

Reference: http://www.lysator.liu.se/c/ANSI-C-grammar-l.html
==============================================================================*/

D     [0-9]
L     [a-zA-Z_]
E     [Ee][+-]?{D}+
FS    (f|F)

%{
#include <stdio.h>
extern FILE *yyin;       // for the file opened by fopen.
extern char *yytext;
extern int yyline;
extern int yylex(void);
%}
%option noyywrap
%%

 /* comments -- don't ask... */
\/\*(\/|\**[^\/\*])*\*+\/     { }
\/\/.*                        { }

 /* key words */
"break"     { return BREAK; }
"connect"   { return CONNECT; }
"continue"  { return CONTINUE; }
"do"        { return DO; }
"else"      { return ELSE; }
"file"      { return FILES; }
"float"     { return FLOAT; }
"for"       { return FOR ; }
"if"        { return IF ; }
"int"       { return INT; }
"job"       { return JOB; }
"length"    { return LENGTH; }
"list"      { return LIST; }
"loadcsv"   { return LOADCSV; }
"map"       { return MAP; }
"main"      { return MAIN; }
"null"      { return NULLS; }
"print"     { return PRINT; }
"push"      { return PUSH; }
"return"    { return RETURN; }
"string"    { return STRING; }
"void"      { return VOID; }
"wait"      { return WAIT; }
"waitall"   { return WAITALL; }
"while"     { return WHILE; }
"mod"       { return MOD; }

{L}({L}|{D})*   { yylval = yytext; return IDENTIFIER; }
{D}+            { yylval = yytext; return CONSTANT; }

 /* exponentials */
{D}+{E}{FS}?            { yylval = yytext; return CONSTANT; }
{D}*"."{D}+({E})?{FS}?  { yylval = yytext; return CONSTANT; }
{D}+"."{D}*({E})?{FS}?  { yylval = yytext; return CONSTANT; }

\"(\\.|[^\\"])*\" { yylval = yytext; return STRING_LITERAL; }

 /* comparison operators */
"&&"      { return AND_OP; }
"||"      { return OR_OP; }
"<="      { return LE_OP; }
">="      { return GE_OP; }
"=="      { return EQ_OP; }
"!="      { return NE_OP; }

"=>"      { return MAPS_TO; }

 /* other characters */
"@"     { return '@'; }
";"     { return ';'; }
"{"     { return '{'; }
"}"     { return '}'; }
","     { return ','; }
"="     { return '='; }
"("     { return '('; }
")"     { return ')'; }
"["     { return '['; }
"]"     { return ']'; }
"."     { return '.'; }
"!"     { return '!'; }
"-"     { return '-'; }
"+"     { return '+'; }
"*"     { return '*'; }
"/"     { return '/'; }
"%"     { return '%'; }
"<"     { return '<'; }
">"     { return '>'; }

[ \t\v\f]+    { }
\n      { yyline++; }
.               { /* ignore bad characters */ }

%%
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=jobthread.cc]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.

#include "jobthread.h"

#include <list>
#include <vector>
#include <map>
#include <iostream>

#include "common.h"
#include "getthread.h"
#include "pigasus-types.h"
#include "sendthread.h"
#include "thread.h"
#include "threadpool.h"

using namespace std;

typedef unsigned int uint;

extern ThreadPool *job_thread_pool;
extern pthread_mutex_t *mutex_job_thread_pool;

extern ThreadPool *send_thread_pool;
extern pthread_mutex_t *mutex_send_thread_pool;

extern ThreadPool *get_thread_pool;
extern pthread_mutex_t *mutex_get_thread_pool;

JobThread::JobThread(Servers &servers_, List &common_, 
                     List &inputs_, List *outputs_) {
  type = kJobThreadType;  

  // Make a local copy of the servers. 
  servers = servers_.Clone();
  // Common is expected to be a list of files. These files will be sent to every
  // remote server.
  common = common_.Clone();
  // Inputs is expected to be a list of lists. Each of these sub-lists will be
  // sent to a single remote server.
  inputs = inputs_.Clone();
  // TODO: what was outputs again?
  outputs = outputs_;
}

JobThread::~JobThread() {
  delete servers;
  delete common;
  delete inputs;
  // Do NOT delete outputs---the user needs that.
}

void JobThread::Run() {
  vector<pair<int, Map*> > sent_ids;
  queue<Map *> return_outputs;
  FILE *p;
  char ignore[20];
  char *err;
  string buffer;
  bool done = false;

  if (inputs->GetSize() == 0) {
    inputs->Append(NewList(0));
  }

  // Send out all the files and remotely execute.
  for (int i = 0; i < inputs->GetSize(); i++) {
    Map *current_server = servers->GetNextServer();
    List *inputs_slice = (List*)(inputs->Get(i));
    Thread *send_thread = new SendThread(*current_server, *common,
                                         *inputs_slice);

    // Add SendThread job to SendThreadPool.
    pthread_mutex_lock(mutex_send_thread_pool);
    int sid = send_thread_pool->AddThread(send_thread);
    pthread_mutex_unlock(mutex_send_thread_pool);

    sent_ids.push_back(pair<int, Map*>(sid, current_server));
    debug("JobThread %d: Added SendThread job %d.\n", id, sid); 
  }

  // All jobs have now been sent, we can begin check if any are finished
  // executing remotely, and if so, we can begin to retrieve the files.
  map<int, bool> already_getting;
  for (int i = 0; i < static_cast<int>(sent_ids.size()); i++) {
    already_getting[sent_ids[i].first] = false;
  }

  list<int> get_ids;
  while (true) {
    bool all_done = true;
    for (int i = 0; i < static_cast<int>(sent_ids.size()); i++) {
      if (already_getting[sent_ids[i].first]) {
        continue;
      }

      // Run the checkdone script for this id (sent_ids[i].first).
      
      Map *server = sent_ids[i].second;
      string user = (*(string*)(server->Get("user")));
      string host = (*(string*)(server->Get("host")));
      
      buffer = "./checkdone.sh " + user + " " + host + " " +
               IntToString(sent_ids[i].first);
      p = popen(buffer.c_str(), "r");
      while (!feof(p)) {
        err = fgets(ignore, sizeof(ignore), p);
        done = (ignore[0] == '1') ? true : false;
      }
      pclose(p);
      if (!done) {
        all_done = false;
        continue;
      }
      
      // This sent job slice is done. We can begin downloading the related files
      // from this server and ID.
      Map *out = new Map();
      out->Set("testkey", "testval");
      debug("Map size (at start) is %d\n", out->GetSize());
      debug("Map address is %p\n", out);
      Thread *get_thread = new GetThread(sent_ids[i].first, *(sent_ids[i].second),
                                         out);

      return_outputs.push(out);
      
      // Add get_thread to GetThreadPool.
      pthread_mutex_lock(mutex_get_thread_pool);
      int gid = get_thread_pool->AddThread(get_thread);
      pthread_mutex_unlock(mutex_get_thread_pool);

      get_ids.push_back(gid);
      already_getting[sent_ids[i].first] = true;
    }

    if (all_done) {
      break;
    } else {
      // Sleep for a while before checking again.
      sleep(kDefaultJobThreadDelay);
    }
  }

  // All GetThreads have been sent. Wait for all the files to be downloaded onto
  // the local machine  
  while (!get_ids.empty()) {
    int get_threads_done = 0; 
    for (list<int>::iterator it = get_ids.begin(); it != get_ids.end(); it++) {
      pthread_mutex_lock(mutex_get_thread_pool);
      debug("Gettings status for %d\n", *it);
      int status = get_thread_pool->GetStatus(*it);
      pthread_mutex_unlock(mutex_get_thread_pool);

      if (status == kThreadStatusDone) {
        get_threads_done++;
      }
    }
    
    if (get_threads_done == static_cast<int>(get_ids.size())) {
      break;
    } else {
      sleep(1);
    }
  }
  
  if (outputs != NULL) {
    while (!return_outputs.empty()) {
      debug("Outputs: %s\n",
            ((string*)((return_outputs.front())->Get("stdout")))->c_str());
      outputs->Append(*(return_outputs.front()));
      return_outputs.pop();
    }
  }
  
  pthread_mutex_lock(mutex_job_thread_pool);
  job_thread_pool->SetStatus(id, kThreadStatusMarkAsDone);
  pthread_mutex_unlock(mutex_job_thread_pool);
  debug("Job thread exiting\n");
  return;
}


\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=jobthread.h]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.
//
// Defines the Job Thread, which is in change of pushes files and getting them
// back.
//
// Remember: compile using -pthread when using threads.

#ifndef JOBTHREADS_H_
#define JOBTHREADS_H_

#include "servers.h"
#include "pigasus-types.h"
#include "thread.h"

using namespace std;

// After all input files have been sent out, this is the number of seconds in
// between checking to see if they're all done.
const int kDefaultJobThreadDelay = 1;

// Threads of this type are for each job. The JobThread is responsible for
// sending each of the input cases out to a machine, checking to see which cases
// are done, marking a job as done if all cases are done, and retrieving the
// files for the done cases. Sending the files is done by spawning SendThreads.
// Retrieving files from remote servers is done by spawning GetThreads.
class JobThread : public Thread {
  Servers *servers;
  List *common;
  List *inputs;
  List *outputs;

 public:
  // 
  JobThread(Servers &servers_, List &common_, List &inputs_, List *outputs_);
  ~JobThread();

  void Run();  
};

#endif  // JOBTHREAD_H_
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=loadcsv.cc]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.

#include "loadcsv.h"

#include <iostream>
#include <fstream>

#include "pigasus-types.h"

using namespace std;

typedef unsigned int uint; // Why is this necessary?

bool LoadCSVIntoVector(string filename, vector< vector<string> > *csv_elements) {
  ifstream csv_file(filename.c_str());

  if (!csv_file.is_open()) {
    return false;
  }

  string line;

  while (!csv_file.eof()) {
    getline(csv_file, line);

    if (line.empty()) {
      continue;
    }

    vector<string> elements;
    string buffer;
    bool in_quotes = false;

    do {
      if (in_quotes) {
        // Handles newlines in CSV elements.
        buffer += '\n';
        if (!csv_file.eof()) {
          getline(csv_file, line);
        } else {
          break;
        }
      }

      for (uint i = 0; i < line.size(); i++) {
        char c = line[i];
        if (c == kQuoteChar) {
          // Quote may end a quoted block, or escape another quote.
          if (in_quotes
              && (i+1 < line.size())
              && (line[i+1] == kQuoteChar)) {
            // Two quotes in a row means one scaped quote.
            buffer += line[i+1];
            i++;
          } else {
            in_quotes = !in_quotes;
            // Embedded quote in the middle: a,b"c"d,e
            if (i > 2
                && (line[i-1] != kSeparator)
                && (i+1 < line.size())
                && (line[i+1] != kSeparator)) {
              buffer += c;
            } 
          }
        } else if (c == kSeparator && !in_quotes) {
          elements.push_back(buffer);
          buffer.clear();
        } else {
          buffer += c;
        }
      }
    } while (in_quotes);

    elements.push_back(buffer);
    csv_elements->push_back(elements);
  }

  csv_file.close();
  return true;
}

List LoadCSV(string filename) {
  vector< vector<string> > csv_elements;
  List csv;

  if (!LoadCSVIntoVector(filename, &csv_elements) || csv_elements.empty()) {
    return csv;  // Returns empty list.
  }
  
  // First row of CSV are the headers. Each element will be subsequently mapped
  // to its corresponding header.
  for (uint i = 1; i < csv_elements.size(); i ++) {
    Map *csv_row = new Map();
   
    for (uint j = 0;
         (j < csv_elements[i].size()) && (j < csv_elements[0].size());
         j++) {
      csv_row->Set(csv_elements[0][j], csv_elements[i][j]);
    }
    csv.Append(*csv_row);
    delete csv_row;
  }

  return csv;   
}
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=loadcsv.h]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.
//
// Implements the LoadCSV function which reads a CSV file into a List of Maps.
// The following implementation is based on that of the Java OpenCSV library:
//   http://opencsv.sourceforge.net/

#ifndef LOADCSV_H_
#define LOADCSV_H_

#include <string>
#include <vector>

#include "pigasus-types.h"

using namespace std;

const char kQuoteChar = '\"';
const char kSeparator = ',';

// Reads a CSV file and tries to load the CSV elements.
bool LoadCSVIntoVector(string filename, vector< vector<string> > *csv_elements);

// Reads a CSV file and returns it in a List of Maps. Each row will be a Map,
// which will be a mapping of the headers to the elements in their corresponding
// columns.
List LoadCSV(string filename);

#endif  // LOADCSV_H_
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=main.cc]
// Main.cc
// This is basically just a wrapper for PigMain()
// It sets up the thread pools, initializes & cleans up SSH keys, etc.

#include <stdlib.h>

#include <iostream>

#include "servers.h"
#include "thread.h"
#include "jobthread.h"
#include "sendthread.h"
#include "threadpool.h"
#include "pigasus-types.h"
#include "push.h"
#include "wait.h"

using namespace std;

// Create threadpools and associated mutexes.
ThreadPool *job_thread_pool;
pthread_mutex_t *mutex_job_thread_pool;

ThreadPool *send_thread_pool;
pthread_mutex_t *mutex_send_thread_pool;

ThreadPool *get_thread_pool;
pthread_mutex_t *mutex_get_thread_pool;

Servers *global_servers;

// We expect PigMain from the translator.
extern void PigMain(List &args);

/*
void PigMain(List &args) {
  cout << "Pig main!" << endl;
  Map m;
  m.Set("host", "localhost");
  m.Set("user", "dummy");
  m.Set("password", "dummypw");
  List my_servers = NewList(m);
  
  //my_servers.Append(Map("user", "dummy"));
  //my_servers.Append(Map("password", "dummypw"));
  
  global_servers->SetServers(my_servers);
  
  List common = NewList(File("foo.bar"));
  // As I recall, inputs is suppose to be a list of lists? --John
  
  List inputs = NewList(NewList(File("pig")))
                + NewList(NewList(File("pigasus-types.o")));
  
  // List inputs;
  // inputs.Append(NewList(File("pig")));
  // inputs.Append(NewList(File("pigasus-types.o")));
  List outputs;
  
  int id = Push("a", common, inputs, NULL);

  Wait(id);

  printf("******* Wait is working!\n");

  // While a thread is still executing, keep all the threadpools alive.
  while (true) {
    bool a = job_thread_pool->HasJobsRunning();
    bool b = send_thread_pool->HasJobsRunning();
    bool c = get_thread_pool->HasJobsRunning();

    if (!(a || b || c)) break;
    sleep(1);
  }
}
*/

int main(int argc, char **argv, char **envp) {
  FILE *p;
  char ignore[20];
  char *err;
  // Initialize global pointers.
  job_thread_pool = new ThreadPool(kJobThreadType);
  mutex_job_thread_pool = new pthread_mutex_t;
  pthread_mutex_init(mutex_job_thread_pool, NULL);

  send_thread_pool = new ThreadPool(kSendThreadType);
  mutex_send_thread_pool = new pthread_mutex_t;
  pthread_mutex_init(mutex_send_thread_pool, NULL);

  get_thread_pool = new ThreadPool(kGetThreadType);
  mutex_get_thread_pool = new pthread_mutex_t;
  pthread_mutex_init(mutex_get_thread_pool, NULL);

  // The post_thread_pool should allow infinite (i.e. INT_MAX) threads.
  // post_thread_pool = new ThreadPool(-1);
  
  global_servers = new Servers();
  
  // Start the threads. They won't actually do anything yet.
  pthread_mutex_lock(mutex_job_thread_pool);
  job_thread_pool->Start();
  pthread_mutex_unlock(mutex_job_thread_pool);

  pthread_mutex_lock(mutex_send_thread_pool);
  send_thread_pool->Start();
  pthread_mutex_unlock(mutex_send_thread_pool);

  pthread_mutex_lock(mutex_get_thread_pool);
  get_thread_pool->Start();
  pthread_mutex_unlock(mutex_get_thread_pool);

  // post_thread_pool->Start();
  
  // Prepare List args for PigMain. Don't skip the first arg (executible name). 
  List args;
  for (int i = 0; i < argc; ++i) {   
    args.Append(string(argv[i]));
  }
  
  // Generates files pig_rsa and pig_rsa.pub
  p = popen("./makekey.sh", "r");
  while (!feof(p)) err = fgets(ignore, sizeof(ignore), p);
  pclose(p);
  
  PigMain(args);

  // All thread pools are now empty, and we are ready to end the program.
  debug("All thread pools are now empty.\n");

  pthread_mutex_destroy(mutex_job_thread_pool);
  pthread_mutex_destroy(mutex_send_thread_pool);
  pthread_mutex_destroy(mutex_get_thread_pool);

  delete mutex_job_thread_pool;
  delete mutex_send_thread_pool;
  delete mutex_get_thread_pool;
  
  // Clean up ssh files. TODO: delete keys on the remote servers.
  p = popen("ssh-add -d pig_rsa", "r");
  while (!feof(p)) err = fgets(ignore, sizeof(ignore), p);
  pclose(p);
  
  p = popen("rm pig_rsa pig_rsa.pub", "r");
  while (!feof(p)) err = fgets(ignore, sizeof(ignore), p);
  pclose(p);

  cout << "Done!" << endl;
  return 0;
}
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=makekey.sh]
#!/usr/bin/env bash
# Author: Jordan Rupprecht

ssh-add -d pig_rsa &> /dev/null
rm pig_rsa* &> /dev/null
ssh-keygen -t rsa -N "" -f pig_rsa &> /dev/null
chmod 0600 pig_rsa* &> /dev/null
ssh-add pig_rsa &> /dev/null
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=pig]
#!/usr/bin/env bash

SRC=`dirname $0`
PWD=`pwd`
PIGOBJS="$SRC/pigasus-types.o $SRC/formatstring.o $SRC/loadcsv.o $SRC/cast.o $SRC/serialization.o $SRC/common.o"
echo "Pig compiler"

PIGOUT="pig.out"

while getopts ":o:" opt; do
  case $opt in
    o)
      PIGOUT="$OPTARG"
      ;;
    *)
      ;;
  esac
done

# Compiling the object files is expensive, so check first
make -C $SRC -q
if [ $? != 0 ]
then
  echo "Building Pigasus object files"
  make -C $SRC -j
fi

# Compiling the grammar is easy, just do it every time...
make -C $SRC/grammar > /dev/null

# ${!#} is the last argument (i.e. source.pig)
$SRC/grammar/pigtrans < ${!#} > /dev/null
if [ $? != 0 ]
then
  echo "Error during parsing; exiting"
  exit 1
fi

cp $SRC/*.sh $PWD

g++ -o $PIGOUT -I$SRC -pthread pigMain.cc $SRC/*.o

for f in $( ls $PWD/*.cc )
do
  if [ `basename $f` == "pigMain.cc" ]
  then
    continue
  fi
  g++ -o `basename $f .cc`.out -I$SRC -pthread $f $PIGOBJS;
done
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=pigasus-types.cc]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.

#include "pigasus-types.h"

#include <iostream>
#include <map>
#include <string>
#include <vector>

using namespace std;

typedef unsigned int uint; // Why is this necessary?

void DeleteElement(string type, void* ptr) {
  if (ptr == NULL) {
    // Trying to delete NULL pointer. Ignore.
    return;
  }

  if (type == kIntType) {
    delete (int*)ptr;
  } else if (type == kFloatType) {
    delete (double*)ptr;
  } else if (type == kStringType) {
    delete (string*)ptr;
  } else if (type == kListType) {
    delete (List*)ptr;
  } else if (type == kMapType) {
    delete (Map*)ptr;
  } else if (type == kFileType) {
    delete (File*)ptr;
  } else {
    // TODO: unknown type inserted; cannot be deleted.
  }
}

List NewList(const int &v) {
  List new_list;
  new_list.Append(v);
  return new_list;
}

List NewList(const double &v) {
  List new_list;
  new_list.Append(v);
  return new_list;
}

List NewList(const string &v) {
  List new_list;
  new_list.Append(v);
  return new_list;
}

List NewList(const List &v) {
  List new_list;
  new_list.Append(v);
  return new_list;
}

List NewList(const Map &v) {
  List new_list;
  new_list.Append(v);
  return new_list;
}

List NewList(const File &v) {
  List new_list;
  new_list.Append(v);
  return new_list;
}

void List::CopyIntoThis(const List &v)  {
  // Empty the current list first.
  for (unsigned int i = 0; i < value.size(); i++) {
    DeleteElement(value[i].first, value[i].second);
  }
  value.clear();
  
  
  // Set this to the specified list.
  List *w = (List*)(&v);
  for (int i = 0; i < w->GetSize(); i++) {
    if (w->GetType(i) == kIntType) {
      Append(*((int*)w->Get(i)));
    } else if (w->GetType(i) == kFloatType) {
      Append(*((double*)w->Get(i)));
    } else if (w->GetType(i) == kStringType) {
      Append(*((string*)w->Get(i)));
    } else if (w->GetType(i) == kListType) {
      Append(*((List*)w->Get(i)));
    } else if (w->GetType(i) == kMapType) {
      Append(*((Map*)w->Get(i)));
    } else if (w->GetType(i) == kFileType) {
      Append(*((File*)w->Get(i)));
    } else {
      // TODO: Unrecognized type found. Ignore for now. Decide what to do.
    }
  }
}

List::List() {
  // Empty constructor.
}

// Overload copy constructor.
List::List(const List &v)  {
  CopyIntoThis(v);
}

// Overload copy operator.
List& List::operator=(const List &v) {
  CopyIntoThis(v);
  return *this;
}

List::~List() {
  for (uint i = 0; i < value.size(); i++) {
    DeleteElement(value[i].first, value[i].second);
  }
}

List* List::Clone() {
  List* copy = new List();

  for (uint i = 0; i < value.size(); i++) {
    if (value[i].first == kIntType) {
      copy->Append(*((int*)value[i].second));
    } else if (value[i].first == kFloatType) {
      copy->Append(*((double*)value[i].second));
    } else if (value[i].first == kStringType) {
      copy->Append(*((string*)value[i].second));
    } else if (value[i].first == kListType) {
      copy->Append(*(((List*)value[i].second)->Clone()));
    } else if (value[i].first == kMapType) {
      copy->Append(*(((Map*)value[i].second)->Clone()));
    } else if (value[i].first == kFileType) {
      copy->Append(*(((File*)value[i].second)->Clone()));
    } else {
      // TODO: unknown type found; cannot be copied.
    }
  }

  return copy;  
}

List List::CloneValue() {
  List copy;

  for (uint i = 0; i < value.size(); i++) {
    if (value[i].first == kIntType) {
      copy.Append(*((int*)value[i].second));
    } else if (value[i].first == kFloatType) {
      copy.Append(*((double*)value[i].second));
    } else if (value[i].first == kStringType) {
      copy.Append(*((string*)value[i].second));
    } else if (value[i].first == kListType) {
      copy.Append(*(((List*)value[i].second)->Clone()));
    } else if (value[i].first == kMapType) {
      copy.Append(*(((Map*)value[i].second)->Clone()));
    } else if (value[i].first == kFileType) {
      copy.Append(*(((File*)value[i].second)->Clone()));
    } else {
      // TODO: unknown type found; cannot be copied.
    }
  }

  return copy;  
}

void* List::Get(int index) {
  if (index >= static_cast<int>(value.size())) {
    return NULL;
  } else {
    return value[index].second;
  }
}

void* List::operator[](const int &index) {
  return Get(index);
}

void List::Append(const int &v) {
  int *t = new int(v);
  value.push_back(pair<string, void*>(kIntType, (void*)t));
}

void List::Append(const double &v) {
  double *t = new double(v);
  value.push_back(pair<string, void*>(kFloatType, (void*)t));
}

void List::Append(const string &v) {
  string *t = new string(v);
  value.push_back(pair<string, void*>(kStringType, (void*)t));
}

void List::Append(const List &v) {
  List *t = ((List*)(&v))->Clone();
  value.push_back(pair<string, void*>(kListType, (void*)t));
}

void List::Append(const Map &v) {
  Map *t = ((Map*)(&v))->Clone();
  value.push_back(pair<string, void*>(kMapType, (void*)t));
}

void List::Append(const File &v) {
  File *t = ((File*)(&v))->Clone();
  value.push_back(pair<string, void*>(kFileType, (void*)t));
}

void List::Set(int index, const int &v) {
  if (index >= GetSize()) {
    // TODO: decide how to handle index-out-of-bounds error.
    return;
  }

  DeleteElement(GetType(index), Get(index));
  int *t = new int(v);
  value[index] = pair<string, void*>(kIntType, (void*)t);
}

void List::Set(int index, const double &v) {
  if (index >= GetSize()) {
    // TODO: decide how to handle index-out-of-bounds error.
    return;
  }

  DeleteElement(GetType(index), Get(index));
  double *t = new double(v);
  value[index] = pair<string, void*>(kFloatType, (void*)t);
}

void List::Set(int index, const string &v) {
  if (index >= GetSize()) {
    // TODO: decide how to handle index-out-of-bounds error.
    return;
  }

  DeleteElement(GetType(index), Get(index));
  string *t = new string(v);
  value[index] = pair<string, void*>(kStringType, (void*)t);
}

void List::Set(int index, const List &v) {
  if (index >= GetSize()) {
    // TODO: decide how to handle index-out-of-bounds error.
    return;
  }

  DeleteElement(GetType(index), Get(index));
  List *t = ((List*)(&v))->Clone();
  value[index] = pair<string, void*>(kListType, (void*)t);
}

void List::Set(int index, const Map &v) {
  if (index >= GetSize()) {
    // TODO: decide how to handle index-out-of-bounds error.
    return;
  }

  DeleteElement(GetType(index), Get(index));
  Map *t = ((Map*)(&v))->Clone();
  value[index] = pair<string, void*>(kMapType, (void*)t);
}

void List::Set(int index, const File &v) {
  if (index >= GetSize()) {
    // TODO: decide how to handle index-out-of-bounds error.
    return;
  }

  DeleteElement(GetType(index), Get(index));
  File *t = ((File*)(&v))->Clone();
  value[index] = pair<string, void*>(kFileType, (void*)t);
}

List List::Concatenate(const List &v) {
  List t;

  // Add items from current List.
  for (int i = 0; i < GetSize(); i++) {
    if (GetType(i) == kIntType) {
      t.Append(*((int*)Get(i)));
    } else if (GetType(i) == kFloatType) {
      t.Append(*((double*)Get(i)));
    } else if (GetType(i) == kStringType) {
      t.Append(*((string*)Get(i)));
    } else if (GetType(i) == kListType) {
      t.Append(*((List*)Get(i)));
    } else if (GetType(i) == kMapType) {
      t.Append(*((Map*)Get(i)));
    } else if (GetType(i) == kFileType) {
      t.Append(*((File*)Get(i)));
    } else {
      // TODO: Unrecognized type found. Ignore for now. Decide what to do.
    }
  }

  // Add items from other List;
  List *w = (List*)(&v);
  for (int i = 0; i < w->GetSize(); i++) {
    if (w->GetType(i) == kIntType) {
      t.Append(*((int*)w->Get(i)));
    } else if (w->GetType(i) == kFloatType) {
      t.Append(*((double*)w->Get(i)));
    } else if (w->GetType(i) == kStringType) {
      t.Append(*((string*)w->Get(i)));
    } else if (w->GetType(i) == kListType) {
      t.Append(*((List*)w->Get(i)));
    } else if (w->GetType(i) == kMapType) {
      t.Append(*((Map*)w->Get(i)));
    } else if (w->GetType(i) == kFileType) {
      t.Append(*((File*)w->Get(i)));
    } else {
      // TODO: Unrecognized type found. Ignore for now. Decide what to do.
    }
  }

  return t;
}

List List::operator+(const List &v) {
  return Concatenate(v);
}

int List::GetSize() {
  return static_cast<int>(value.size());
}

string List::GetType(int index) {
  if (index >= static_cast<int>(value.size())) {
    return kNullType;
  } else {
    return value[index].first;
  }
}

void Map::CopyIntoThis(const Map &v) {
  // Empty out this one first.
  for (map<string, pair<string, void*> >::iterator it = value.begin();
       it != value.end();
       it++) {
    DeleteElement(it->second.first, it->second.second);
  }
  value.clear();

  // Copy the specified map into this one.
  Map *w = (Map*)(&v);
  List keys = w->GetKeys();

  for (int i = 0; i < keys.GetSize(); i++) {
    string key = *((string*)(keys[i]));
    string type = w->GetType(key);

    if (type == kIntType) {
      Set(key, *((int*)(w->Get(key))));
    } else if (type == kFloatType) {
      Set(key, *((double*)(w->Get(key))));
    } else if (type == kStringType) {
      Set(key, *((string*)(w->Get(key))));
    } else if (type == kListType) {
      Set(key, *(((List*)(w->Get(key)))->Clone()));
    } else if (type == kMapType) {
      Set(key, *(((Map*)(w->Get(key)))->Clone()));
    } else if (type == kFileType) {
      Set(key, *(((File*)(w->Get(key)))->Clone()));
    } else {
      // TODO: unknown type found; cannot be copied.
    }
  }
}

Map::Map() {
  // Empty constructor.
}

Map::Map(const Map& v) {
  CopyIntoThis(v);
}

Map& Map::operator=(const Map& v) {
  CopyIntoThis(v);
  return *this;
}

Map::Map(const string &key, const int &v) {
  Set(key, v);
}

Map::Map(const string &key, const double &v) {
  Set(key, v);
}

Map::Map(const string &key, const string &v) {
  Set(key, v);
}

Map::Map(const string &key, const List &v) {
  Set(key, v);
}

Map::Map(const string &key, const Map &v) {
  Set(key, v);
}

Map::Map(const string &key, const File &v) {
  Set(key, v);
}

Map::~Map() {
  for (map< string, pair<string, void*> >::iterator it = value.begin();
       it != value.end();
       it++) {
    DeleteElement(it->second.first, it->second.second);
  }
}

Map* Map::Clone() {
  Map *copy = new Map();

  for (map< string, pair<string, void*> >::iterator it = value.begin();
       it != value.end();
       it++) {
    if (it->second.first == kIntType) {
      copy->Set(it->first, *((int*)it->second.second));
    } else if (it->second.first == kFloatType) {
      copy->Set(it->first, *((double*)it->second.second));
    } else if (it->second.first == kStringType) {
      copy->Set(it->first, *((string*)it->second.second));
    } else if (it->second.first == kListType) {
      copy->Set(it->first, *(((List*)it->second.second)->Clone()));
    } else if (it->second.first == kMapType) {
      copy->Set(it->first, *(((Map*)it->second.second)->Clone()));
    } else if (it->second.first == kFileType) {
      copy->Set(it->first, *(((File*)it->second.second)->Clone()));
    } else {
      // TODO: unknown type found; cannot be copied.
    }
  }

  return copy;
}

Map Map::CloneValue() {
  Map copy;

  for (map< string, pair<string, void*> >::iterator it = value.begin();
       it != value.end();
       it++) {
    if (it->second.first == kIntType) {
      copy.Set(it->first, *((int*)it->second.second));
    } else if (it->second.first == kFloatType) {
      copy.Set(it->first, *((double*)it->second.second));
    } else if (it->second.first == kStringType) {
      copy.Set(it->first, *((string*)it->second.second));
    } else if (it->second.first == kListType) {
      copy.Set(it->first, *(((List*)it->second.second)->Clone()));
    } else if (it->second.first == kMapType) {
      copy.Set(it->first, *(((Map*)it->second.second)->Clone()));
    } else if (it->second.first == kFileType) {
      copy.Set(it->first, *(((File*)it->second.second)->Clone()));
    } else {
      // TODO: unknown type found; cannot be copied.
    }
  }

  return copy;
}

void* Map::Get(string key) {
  if (value.find(key) == value.end()) {
    return NULL;
  } else {
    return value[key].second;
  }
}

void* Map::operator[](const string &key) {
  return Get(key);
}

void Map::Set(const string &key, const int &v) {
  DeleteElement(GetType(key), Get(key));

  int *t = new int(v);
  value[key] = pair<string, void*>(kIntType, (void*)t);
}

void Map::Set(const string &key, const double &v) {
  DeleteElement(GetType(key), Get(key));

  double *t = new double(v);
  value[key] = pair<string, void*>(kFloatType, (void*)t);
}

void Map::Set(const string &key, const string &v) {
  DeleteElement(GetType(key), Get(key));

  string *t = new string(v);
  value[key] = pair<string, void*>(kStringType, (void*)t);
}

void Map::Set(const string &key, const List &v) {
  DeleteElement(GetType(key), Get(key));

  List *t = ((List*)(&v))->Clone();
  value[key] = pair<string, void*>(kListType, (void*)t);
}

void Map::Set(const string &key, const Map &v) {
  DeleteElement(GetType(key), Get(key));

  Map *t = ((Map*)(&v))->Clone();
  value[key] = pair<string, void*>(kMapType, (void*)t);
}

void Map::Set(const string &key, const File &v) {
  DeleteElement(GetType(key), Get(key));

  File *t = ((File*)(&v))->Clone();
  value[key] = pair<string, void*>(kFileType, (void*)t);
}

Map Map::Concatenate(const Map &v) {
  Map t;

  // Add items from current Map.
  for (map< string, pair<string, void*> >::iterator it = value.begin();
       it != value.end();
       it++) {
    if (it->second.first == kIntType) {
      t.Set(it->first, *((int*)it->second.second));
    } else if (it->second.first == kFloatType) {
      t.Set(it->first, *((double*)it->second.second));
    } else if (it->second.first == kStringType) {
      t.Set(it->first, *((string*)it->second.second));
    } else if (it->second.first == kListType) {
      t.Set(it->first, *(((List*)it->second.second)->Clone()));
    } else if (it->second.first == kMapType) {
      t.Set(it->first, *(((Map*)it->second.second)->Clone()));
    } else if (it->second.first == kFileType) {
      t.Set(it->first, *(((File*)it->second.second)->Clone()));
    } else {
      // TODO: unknown type found; cannot be copied.
    }
  }

  // Add items from other Map.
  Map *w = (Map*)(&v);
  List keys = w->GetKeys();

  for (int i = 0; i < keys.GetSize(); i++) {
    string key = *((string*)(keys[i]));
    string type = w->GetType(key);

    if (type == kIntType) {
      t.Set(key, *((int*)(w->Get(key))));
    } else if (type == kFloatType) {
      t.Set(key, *((double*)(w->Get(key))));
    } else if (type == kStringType) {
      t.Set(key, *((string*)(w->Get(key))));
    } else if (type == kListType) {
      t.Set(key, *(((List*)(w->Get(key)))->Clone()));
    } else if (type == kMapType) {
      t.Set(key, *(((Map*)(w->Get(key)))->Clone()));
    } else if (type == kFileType) {
      t.Set(key, *(((File*)(w->Get(key)))->Clone()));
    } else {
      // TODO: unknown type found; cannot be copied.
    }
  }

  return t;
}

Map Map::operator+(const Map &v) {
  return Concatenate(v);
}

int Map::GetSize() {
  return static_cast<int>(value.size());
}

string Map::GetType(const string &key) {
  if (value.find(key) == value.end()) {
    return kNullType;
  } else {
    return value[key].first;
  }
}

List Map::GetKeys() {
  List t;

  for (map<string, pair<string, void*> >::iterator it = value.begin();
       it != value.end();
       it++) {
    t.Append(it->first);
  }

  return t;
}

File::File() {
  // Empty constructor.
}

File::File(string filename_) {
  filename = filename_;
}

File::~File() {
  // Empty destructor.
}

File* File::Clone() {
  File *copy = new File(filename);
  return copy;
}

File File::CloneValue() {
  File copy(filename);
  return copy;
}

string File::Filename() {
  return filename;
}



/*
void Test(List a) {
  for (int i = 0; i < a.GetSize(); i++) {
    printf("%s\n", ((string*)a[i])->c_str());
  }
}

void Test(Map a) {
  List k = a.GetKeys();
  for (int i = 0; i < k.GetSize(); i++) {
    string key = *((string*)(k[i]));
    printf("%s -> %s\n", key.c_str(), ((string*)a[key])->c_str());
  }
}

void Test(File a) {
  printf("%s\n", a.Filename().c_str());
}

int main() {
  List aa = NewList("bob") + NewList("bill");
  Test(aa);

  Map bb = Map("a", "b") + Map("c", "d") + Map("e", "f");
  Test(bb);

  File cc = File("a.out");
  Test(cc);

  return 0;
}
*/
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=pigasus-types.h]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.
//
// Defines the types supported by Pigasus. Particularly the classes representing
// the types for storage into Pigasus lists and maps. The following classes are
// defined:
//   List
//   Map
//   File

#ifndef PIGASUS_TYPES_H_
#define PIGASUS_TYPES_H_

#include <map>
#include <string>
#include <vector>

using namespace std;

// String representations of the types. That is, this is what is returned after
// a user uses the Pigasus 'typeof' expression.
const string kIntType = "int";
const string kFloatType = "float";
const string kStringType = "string";
const string kListType = "list";
const string kMapType = "map";
const string kFileType = "file";
const string kNullType = "null";

// Given a pointer and the type as a string, delete it from memory. This
// includes the Pigasus specific types also.
void DeleteElement(string type, void* ptr);

class List;
class Map;
class File;

List NewList(const int &v);
List NewList(const double &v);
List NewList(const string &v);
List NewList(const List &v);
List NewList(const Map &v);
List NewList(const File &v);

// Represents a Pigasus list, stored as a vector of pairs, where the first
// element is the type (stored as a string) and the second is the pointer to
// the actual value. You may declare:
//   List my_list;
// You may also use:
//   List *my_list = new List();
// And remember to clean up after yourself:
//   delete my_list;
class List {
  // Elements of the list are stored as pairs, with the first being the type,
  // and the second being a pointer to the element stored in memory.
  vector< pair<string, void*> > value;

  // Copy.
  void CopyIntoThis(const List &v);

 public:
  // Empty constructor.
  List();
  // Overload copy constructor.
  List(const List &v);
  // Overload copy operator.
  List& operator=(const List &v);
  
  // Destructor will also go through the values and delete everything in the
  // List (recursively if necessary).
  ~List();

  // Creates a new List in memory with copies (not just pointers) of the
  // elements.
  List* Clone();
  // Returns a new List by value with the same values as this.
  List CloneValue();

  // Returns a pointer to the element stored at the specified index. 
  void* Get(int index);
  // Operator associated with Get. So for List v, v[1] is equivalent to Get(1).
  void* operator[](const int &index);

  // Appends an element to the end of the list. Remember to pass the reference
  // for Pigasus-specific types.
  void Append(const int &v);
  void Append(const double &v);
  void Append(const string &v);
  void Append(const List &v);
  void Append(const Map &v);
  void Append(const File &v);
  
  // Replaces the element at the given index. Remember to pass the reference for
  // Pigasus-specific types. TODO: decide how to handle the error if the
  // specified index is out of bounds. Also, the existing element stored at
  // the index is deleted. (Since it's a 'local' copy, it shouldn't affect
  // anything else).
  void Set(int index, const int &v);
  void Set(int index, const double &v);
  void Set(int index, const string &v);
  void Set(int index, const List &v);
  void Set(int index, const Map &v);
  void Set(int index, const File &v);
  
  // Returns a new List which contains the elements of the current List, plus
  // the elements of a specified List at the end.
  List Concatenate(const List &w);
  // Operator associated with Concatenate. So for Lists a, b, c, the following are
  // equivalent: c = a + b and c = a.Concatenate(b).
  List operator+(const List &v);

  // Returns number of objects in list.
  int GetSize();

  // Returns the type of the element stored at the index.
  string GetType(int index);
};

// Represents a Pigasus map, stored as a map mapping strings to pairs where the
// first elements is the type of the value (stored as a string) and the second
// element is a pointer to the value itself. If you use a pointer, please clean
// up afterwards:
//   Map *my_map = new Map();
//   delete my_map;
// Of course, you could simply declare:
//   Map my_map;
class Map {
  // Elements of the Map are stored as C++ maps, which map a string key to a
  // pair, with the first being the type and the second being a pointer to the
  // element stored in memory.
  map< string, pair<string, void*> > value;

  // Copy.
  void CopyIntoThis(const Map &v);

 public:
  // Empty constructor.
  Map();
  // Overload copy constructor.
  Map(const Map &v);
  // Overload copy operator.
  Map& operator=(const Map &v);

  // Maps with an initial element.
  Map(const string &key, const int &v);
  Map(const string &key, const double &v);
  Map(const string &key, const string &v);
  Map(const string &key, const List &v);
  Map(const string &key, const Map &v);
  Map(const string &key, const File &v);
  // Deletes all elements in Map, recursively if necessary.
  ~Map();

  // Returns a pointer to a new Map in memory, with copies of the current Map's
  // elements.
  Map* Clone();
  // Return a copy of the current Map by value.
  Map CloneValue();

  // Returns a pointer to the element stored at key.
  void* Get(string key);
  // Associated with Get.
  void* operator[](const string &key);

  // Sets the element at key. Replaces an existing element if it exists (and the
  // existing element is deleted from memory). Like List, remember to pass in
  // the reference if using Pigasus-specific types.
  void Set(const string &key, const int &v);
  void Set(const string &key, const double &v);
  void Set(const string &key, const string &v);
  void Set(const string &key, const List &v);
  void Set(const string &key, const Map &v);
  void Set(const string &key, const File &v);  

  // Returns a new Map with all the mappings of the current Map, plus all the
  // mappings of the specified Map. The specified Map will be given preference
  // when keys exist in both the current and specified Maps.
  Map Concatenate(const Map &v);
  // Operator associated with concatenate.
  Map operator+(const Map &v);

  // Returns the number of objects in the map.
  int GetSize();

  // Returns the type of the element at key.
  string GetType(const string &key);

  // Returns a List of the keys.
  List GetKeys();
};

// Represents a Pigasus file. Simply a string, indicating the filename.
// Deconstructor for file doesn't do much, but is kept for consistency. 
// TODO: Eventually, it may be necessary that filename be converted into the
// absolute path.
class File {
  string filename;

 public:
  File();
  File(string filename_);
  ~File();

  File* Clone();
  File CloneValue();

  string Filename();
};

#endif  // PIGASUS_TYPES_H_

\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=push.cc]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.

#include "push.h"

#include <string>

#include "common.h"
#include "jobthread.h"
#include "pigasus-types.h"
#include "threadpool.h"

using namespace std;

extern ThreadPool *job_thread_pool;
extern pthread_mutex_t *mutex_job_thread_pool;

extern Servers *global_servers;

int Push(string job_name, List common, List inputs, List *outputs) {
  string job_executable = job_name + ".out";
  List common_with_job = NewList(File(job_executable)) + common;

  // debug("Push: %d, %d\n", common_with_job.GetSize(), inputs.GetSize());
  for (int i = 0; i < common.GetSize(); i++) {
    debug("--Push: %s\n", ((string*)common[i])->c_str());
  }
  
  // Add job_executable to common.
  Thread *j = new JobThread(*global_servers, common_with_job, inputs, outputs);
  // Add job!
  pthread_mutex_lock(mutex_job_thread_pool);
  int job_id = job_thread_pool->AddThread(j);
  pthread_mutex_unlock(mutex_job_thread_pool);  

  return job_id;
}

\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=push.h]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.
//
// Implements the Pigasus push keyword.

#ifndef PUSH_H_ 
#define PUSH_H_

#include <string>

#include "pigasus-types.h"

using namespace std;

int Push(string job_name, List common, List inputs, List *outputs);

#endif  // PUSH_H_
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=push.sh]
#!/usr/bin/env bash
# Author: Jordan Rupprecht

ssh -nT $1@$2 "cd pig-$3; ./runjob.sh $4 $5 < /dev/null >& /dev/null" &
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=runjob.sh]
#!/usr/bin/env bash
# Author: Jordan Rupprecht

PWD=`pwd`
echo `hostname -f` > host
rm -f done &> /dev/null
touch now
sleep 1
./$1 $2 > stdout 2> stderr
CHANGED=`find . -newer now -type f -printf '%P\n' | tr "\n" " "`
eval "touch done"
tar -zcf $PWD/changed.tar.gz $CHANGED
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=sendkey.sh]
#!/usr/bin/env expect
# Author: Jordan Rupprecht
# USAGE - sendkey.sh USER HOST PASSWORD KEYCONTENTS
# e.g. sendkey.sh jordan clic.cs.columbia.edu passw0rd `cat id_rsa.pub`

spawn ssh -q -l [lindex $argv 0] [lindex $argv 1] "mkdir -p ~/.ssh &>/dev/null; chmod 700 ~/.ssh; echo \"[lrange $argv 3 5]\" >> ~/.ssh/authorized_keys; chmod 644 ~/.ssh/authorized_keys"
expect *word:
send "[lindex $argv 2]\n"
expect
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=sendthread.cc]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.

#include "sendthread.h"

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>

#include <sstream>

#include "common.h"
#include "pigasus-types.h"
#include "serialization.h"
#include "threadpool.h"

using namespace std;

extern ThreadPool *send_thread_pool;
extern pthread_mutex_t *mutex_send_thread_pool;

SendThread::SendThread(Map &server_, List &files_, List &inputs_) {
  type = kSendThreadType;
  server = &server_;
  files = &files_;
  inputs = &inputs_;
}

SendThread::~SendThread() {}

void SendThread::Run() {
  FILE *p;
  char ignore[20];
  char *err;
  string buffer;
  string user = (*(string*)(server->Get("user")));
  debug("user is %s\n", user.c_str());
  string host = (*(string*)(server->Get("host")));
  
  debug("SendThread %d: Starting (user: %s, host: %s).\n", id, user.c_str(),
        host.c_str());
  
  // Create pigasus directory  
  debug("SendThread %d: Creating remote directory for Pigasus use.\n", id);
  buffer = "ssh -q " + user + "@" + host + " mkdir -p pig-" + IntToString(id);
  p = popen(buffer.c_str(), "r");
  while (!feof(p)) err = fgets(ignore, sizeof(ignore), p);
  pclose(p);

  // Serialize inputs.
  string serialized_inputs = Serialize(*inputs);
  string serialized_filename = "serialized-inputs-" + IntToString(id) + ".dat";
  if (!WriteStringToFile(serialized_filename, serialized_inputs)) {
    // TODO: Handle error. Problem writing file.
    debug("SendThread %d: Error writing serialized inputs file %s.\n",
          id, serialized_filename.c_str());
  }
  
  // Format string to SCP files over
  buffer = "scp runjob.sh " + serialized_filename;
  debug("There are %d files to send\n", files->GetSize());
  for (int i = 0; i < files->GetSize(); ++i) {
    buffer += " " + (*((string*)files->Get(i)));
  }
  for (int i = 0; i < inputs->GetSize(); ++i) {
    if (inputs->GetType(i) == kFileType) {
      buffer += " " + ((File*)(inputs->Get(i)))->Filename();
    }
  }

  // TODO: add the pigasus job executable...
  // I think this should be done in the JobThread? --John

  // SCP files
  debug("SendThread %d: Sending files...\n", id);
  buffer = buffer + " " + user + "@" + host + ":pig-" + IntToString(id);
  debug("SendThread %d: %d files and %d inputs.\n", id, files->GetSize(),
        inputs->GetSize());
  debug("SendThread %d: buffer: %s.\n", id, buffer.c_str());
  p = popen(buffer.c_str(), "r");
  if (!p) {
    debug("ERROR: cannot open pipe\n");
  }
  while (!feof(p)) {
    err = fgets(ignore, sizeof(ignore), p);
    debug("SendThread %d: ignore: %s\n", id, err);
  }
  pclose(p);

  // Delete local copy of the serialized inputs.
  if (remove(serialized_filename.c_str()) != 0) {
    // TODO: Handle error. Problem deleting serialized inputs file.
    debug("SendThread %d: Error deleting serialized inputs %s.\n", 
          id, serialized_filename.c_str());
  }
  
  // Start execution
  string job_executable = *((string*)files->Get(0));
  buffer = "./push.sh " + user + " " + host + " " + IntToString(id) + " "
    + job_executable + " " + serialized_filename;
  debug("Buffer is %s\n", buffer.c_str());
  p = popen(buffer.c_str(), "r");
  while (!feof(p)) err = fgets(ignore, sizeof(ignore), p);
  pclose(p);
  
  // Set status to done  
  pthread_mutex_lock(mutex_send_thread_pool);
  send_thread_pool->SetStatus(id, kThreadStatusMarkAsDone);
  pthread_mutex_unlock(mutex_send_thread_pool);
}
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=sendthread.h]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.
//
// Defines the SendThread class, which is used for sending files and
// starting execution of tasks

#ifndef SENDTHREAD_H_
#define SENDTHREAD_H_

#include "thread.h"
#include "pigasus-types.h"

class SendThread : public Thread {
  Map *server;
  List *files;
  List *inputs;
  
 public:
  SendThread(Map &server_, List &files_, List &inputs_);
  ~SendThread();
  void Run();
};

#endif // SENDTHREAD_H_
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=serialization.cc]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.

#include "serialization.h"

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <sstream>
#include <string>

#include "pigasus-types.h"

using namespace std;

string AbbreviateType(string type) {
  return type.substr(0, kTypeAbbreviationLength);
}

string FormatSize(int size) {
  char buffer[15];
  sprintf(buffer, "%010d", size);
  return string(buffer);
}

string GetNextType(string buffer) {
  return buffer.substr(0, kTypeAbbreviationLength);
}

string Serialize(int v)  {
  string buffer;
  stringstream stream;
  stream << v;
  buffer = stream.str();
  return AbbreviateType(kIntType) + FormatSize(buffer.size()) + buffer;
}

int UnserializeInt(string &buffer) {
  string int_str = UnserializeString(buffer);
  int i;
  sscanf(int_str.c_str(), "%d", &i);
  return i;
}

string Serialize(double v) {
  string buffer;
  stringstream stream;
  stream << v;
  buffer = stream.str();
  return AbbreviateType(kFloatType) + FormatSize(buffer.size()) + buffer;
}

double UnserializeFloat(string &buffer) {
  string float_str = UnserializeString(buffer);
  double i;
  sscanf(float_str.c_str(), "%lf", &i);
  return i;
}

string Serialize(string v) {
  return AbbreviateType(kStringType) + FormatSize(v.size()) + v;
}

string UnserializeString(string &buffer) {
  string type = buffer.substr(0, kTypeAbbreviationLength);
  int size = atoi(buffer.substr(kTypeAbbreviationLength,
                                kFormattedSizeLength).c_str());
  string res = buffer.substr(kTypeAbbreviationLength + kFormattedSizeLength,
                             size);

  int removed_length = kTypeAbbreviationLength + kFormattedSizeLength + size;
  
  buffer = buffer.substr(removed_length, buffer.size() - removed_length + 1);
  return res;
}

string Serialize(List &v) {
  string buffer;
  buffer = AbbreviateType(kListType) + FormatSize(v.GetSize());
  for (int i = 0; i < v.GetSize(); i++) {
    if (v.GetType(i) == kIntType) {
      buffer += Serialize(*((int*)v.Get(i)));
    } else if (v.GetType(i) == kFloatType) {
      buffer += Serialize(*((double*)v.Get(i)));
    } else if (v.GetType(i) == kStringType) {
      buffer += Serialize(*((string*)v.Get(i)));
    } else if (v.GetType(i) == kListType) {
      buffer += Serialize(*((List*)v.Get(i)));
    } else if (v.GetType(i) == kMapType) {
      buffer += Serialize(*((Map*)v.Get(i)));
    } else if (v.GetType(i) == kFileType) {
      buffer += Serialize(*((File*)v.Get(i)));
    } else {
      // Do nothing.
    }
  }

  return buffer;
}

List* UnserializeList(string &buffer) {
  string type = buffer.substr(0, kTypeAbbreviationLength);
  int size = atoi(buffer.substr(kTypeAbbreviationLength,
                                kFormattedSizeLength).c_str());
  
  int removed_length = kTypeAbbreviationLength + kFormattedSizeLength;  
  buffer = buffer.substr(removed_length, buffer.size() - removed_length + 1);

  List *res = new List();
  for (int i = 0; i < size; i++) {
    string type = GetNextType(buffer);
    if (type == AbbreviateType(kIntType)) {
      res->Append(UnserializeInt(buffer));
    } else if (type == AbbreviateType(kFloatType)) {
      res->Append(UnserializeFloat(buffer));
    } else if (type == AbbreviateType(kStringType)) {
      res->Append(UnserializeString(buffer));
    } else if (type == AbbreviateType(kListType)) {
      List *uns = (UnserializeList(buffer));
      res->Append(*uns);
      delete uns;
    } else if (type == AbbreviateType(kMapType)) {
      Map *uns = (UnserializeMap(buffer));
      res->Append(*uns);
      delete uns;
    } else if (type == AbbreviateType(kFileType)) {
      File *uns = (UnserializeFile(buffer));
      res->Append(*uns);
      delete uns;
    } else {
      // Do nothing.
    }
  }
  return res;
}

string Serialize(Map &v) {
  string buffer;
  buffer = AbbreviateType(kMapType) + FormatSize(v.GetSize());

  List keys = v.GetKeys();
  for (int i = 0; i < keys.GetSize(); i++) {
    string key = *((string*)keys.Get(i));
    buffer += Serialize(key);

    if (v.GetType(key) == kIntType) {
      buffer += Serialize(*((int*)v.Get(key)));
    } else if (v.GetType(key) == kFloatType) {
      buffer += Serialize(*((double*)v.Get(key)));
    } else if (v.GetType(key) == kStringType) {
      buffer += Serialize(*((string*)v.Get(key)));
    } else if (v.GetType(key) == kListType) {
      buffer += Serialize(*((List*)v.Get(key)));
    } else if (v.GetType(key) == kMapType) {
      buffer += Serialize(*((Map*)v.Get(key)));
    } else if (v.GetType(key) == kFileType) {
      buffer += Serialize(*((File*)v.Get(key)));
    } else {
      // Do nothing.
    }
  }
  return buffer;
}

Map* UnserializeMap(string &buffer) {
  string type = buffer.substr(0, kTypeAbbreviationLength);
  int size = atoi(buffer.substr(kTypeAbbreviationLength,
                                kFormattedSizeLength).c_str());
  
  int removed_length = kTypeAbbreviationLength + kFormattedSizeLength;  
  buffer = buffer.substr(removed_length, buffer.size() - removed_length + 1);

  Map *res = new Map();
  for (int i = 0; i < size; i++) {
    string key = UnserializeString(buffer);
    string type = GetNextType(buffer);
    if (type == AbbreviateType(kIntType)) {
      res->Set(key, UnserializeInt(buffer));
    } else if (type == AbbreviateType(kFloatType)) {
      res->Set(key, UnserializeFloat(buffer));
    } else if (type == AbbreviateType(kStringType)) {
      res->Set(key, UnserializeString(buffer));
    } else if (type == AbbreviateType(kListType)) {
      List *uns = (UnserializeList(buffer));
      res->Set(key, *uns);
      delete uns;
    } else if (type == AbbreviateType(kMapType)) {
      Map *uns = (UnserializeMap(buffer));
      res->Set(key, *uns);
      delete uns;
    } else if (type == AbbreviateType(kFileType)) {
      File *uns = (UnserializeFile(buffer));
      res->Set(key, *uns);
      delete uns;
    } else {
      // Do nothing.
    }
  }
  return res;
}

string Serialize(File &v) {
  return AbbreviateType(kFileType) + FormatSize(v.Filename().size())
    + v.Filename();
}

File* UnserializeFile(string &buffer) {
  string filename = UnserializeString(buffer);
  File *file = new File(filename);
  return file;
}

/*
int main() {
  Map m;
  m.Set("hello", "world");

  List l;
  l.Append(1);
  l.Append("hi");
  l.Append(m);
  l.Append(((File)"bob.txt"));

  string s = Serialize(l);
  WriteStringToFile("test.out", s);
  
  string ss;
  ReadStringFromFile("test.out", ss);

  List *ll = UnserializeList(ss);
  Map *mm = (Map*)ll->Get(2);
  File *ff = (File*)ll->Get(3);

  cout << *((int*)ll->Get(0)) << endl;
  cout << *((string*)ll->Get(1)) << endl;
  cout << *((string*)mm->Get("hello")) << endl;
  cout << ff->Filename() << endl;
  
  delete ll;

  return 0;
}
*/
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=serialization.h]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.
//
// Defines functions for serializing Pigasus types and unserializing them.
//
// Please NOTE: after you unserialize a list/map/file, make sure you delete that
// pointer when you're done!

#ifndef SERIALIZATION_H_ 
#define SERIALIZATION_H_

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <string>

#include "pigasus-types.h"

using namespace std;

// Number of characters needed to represent a type in serialized form.
const int kTypeAbbreviationLength = 3;
// Number of characters needed to represent a size in serialized form.
const int kFormattedSizeLength = 10;

// Return the first kTypeAbbreviationLength characters of the type.
string AbbreviateType(string type);

// Pad the size with 0's until the string is size kFormattedSizeLength.
string FormatSize(int size);

// Peek ahead in the buffer to see what the next type is.
string GetNextType(string buffer);

// Serialize / unserialize ints.
string Serialize(int v);
int UnserializeInt(string &buffer);

// Serialize / unserialize floats.
string Serialize(double v);
double UnserializeFloat(string &buffer);

// Serialize / unserialize strings.
string Serialize(string v);
string UnserializeString(string &buffer);

// Serialize / unserialize lists.
string Serialize(List &v);
List* UnserializeList(string &buffer);

// Serialize / unserialize maps.
string Serialize(Map &v);
Map* UnserializeMap(string &buffer);

// Serialize / unserialize files.
string Serialize(File &v);
File* UnserializeFile(string &buffer);

#endif  // SERIALIZATION_H_
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=servers.cc]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.

#include "servers.h"

#include <stdio.h>

#include "common.h"

using namespace std;
  
Servers::Servers() {
  next_server_index = 0;
  servers = NULL;
}

Servers::~Servers() {
  if (servers) delete servers;
}

Servers::Servers(const Servers& s) {
  next_server_index = 0;
  if (s.servers) {
    servers = s.servers->Clone();
    has_password.resize(servers->GetSize());
    for (int i = 0; i < static_cast<int>(has_password.size()); i++) {
      has_password[i] = false;
    }
  }
}

Servers* Servers::Clone() {
  return new Servers(*this);
}

void Servers::SetServers(List &servers_) {
  if (servers) delete servers;
  servers = servers_.Clone();
}

Map* Servers::GetNextServer() {
  if (!servers || !servers->GetSize()) {
    // Servers not set!!!
    return NULL;
  }
  // Basic round-robin scheduling. TODO: Weighted round robin.
  Map* ret = (Map*)((*servers)[next_server_index]);

  // Send password here, if we haven't already for this server.
  if (!has_password[next_server_index]) {
    FILE *p;
    char ignore[20];
    char *err;
    string buffer;
    string user = (*(string*)(ret->Get("user")));
    string host = (*(string*)(ret->Get("host")));
    string password = (*(string*)(ret->Get("password")));
  
    debug("Sending password (user: %s, host: %s).\n", user.c_str(),
          host.c_str());
  
    // Send SSH keys.
    buffer = "./sendkey.sh " + user + " " + host + " " + password 
      + " `cat pig_rsa.pub` &> /dev/null" ;
    p = popen(buffer.c_str(), "r");
    while (!feof(p)) err = fgets(ignore, sizeof(ignore), p);
    pclose(p);
    has_password[next_server_index] = true;
  }

  next_server_index = (next_server_index + 1) % servers->GetSize();

  return ret;
}
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=servers.h]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.
//
// The Servers class. This manages server connections/logins/etc.

#ifndef SERVERS_H_
#define SERVERS_H_

#include "pigasus-types.h"

using namespace std;

// Server class
class Servers {
  // List of servers. This is a List of Maps, where each Map looks something
  // like ["host" => "hostname.com", "user" => "mylogin"]
  List *servers;
  vector<bool> has_password;
  
  // Index used for round-robin scheduling
  int next_server_index;
 
 public:
  // Constructor.
  Servers();
  // Destructor.
  ~Servers();
  // Copy constructor
  Servers(const Servers& s);
  // Clone method
  Servers* Clone();
  // Declare servers with a List
  void SetServers(List &servers_);
  // Get the next server to use
  Map *GetNextServer();
};

#endif  // SERVERS_H_
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=system.h]
// Defines the System caller, which is basically a wrapper for system()
#ifndef SYSTEM_H_
#define SYSTEM_H_

#include <stdlib.h>

#include <string>

using namespace std;

static inline int System(const string cmd) {
  return system(cmd.c_str());
}

static inline int System(const char *cmd) {
  return system(cmd);
}

#endif // SYSTEM_H_

\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=thread.h]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.
//
// Defines the parent Thread class. All threads should be derived from this, 
// which allows them to execute in a separate process.
//
// Remember: compile using -pthread when using threads.

#ifndef THREAD_H_
#define THREAD_H_

#include <stdlib.h>
#include <pthread.h>
#include <time.h>
#include <unistd.h>

#include "common.h"

using namespace std;

const int kNullThreadType = 0;
const int kJobThreadType = 1;
const int kSendThreadType = 2;
const int kGetThreadType = 3;
const int kPollThreadType = 4;

// Base class for threads. All classes that are to execute in separate threads
// should be derived from this one.
class Thread {
  pthread_t thread;

  // Run the thread.
  static void *ThreadFunction(void *t) { 
    ((Thread*)t)->Run();
    return (void *)pthread_self(); 
  } 
  
 protected:
  // Thread ID for usage with ThreadPool.
  int id; 
  int type;

 public:
  // Constructor sets the ID.
  Thread() { }

  // Empty virtual destructor.
  virtual ~Thread() { }

  // Actual job of the thread. Must be overloaded.
  virtual void Run() { }

  // Start thread execution with specified ID.   
  int Start(int id_) {
    id = id_;

    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

    int tid = pthread_create(&thread, NULL, Thread::ThreadFunction,
                             (void*)this);

    pthread_attr_destroy(&attr);
    return tid;
  }
  // Start thread execution with default ID -1.
  int Start() {
    return Start(-1);
  }

  // Wait until Run() finishes doing its thing.
  int Wait() {
    return pthread_join(thread, NULL);
  }

  // Return the type ID of this thread.
  int GetType() {
    return type;
  }
};

#endif  // THREAD_H_
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=threadpool.cc]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.

#include "threadpool.h"

#include <limits.h>
#include <pthread.h>
#include <stdlib.h>
#include <time.h>
#include <unistd.h>
#include <sched.h>

#include "common.h"

using namespace std;

ThreadPool::ThreadPool(int type_, int size_, int delay_) {
  // Initialize status variables.
  next_thread_id = type_ * 10000;
  num_running_threads = 0;

  // Initialize some parameters.
  type = type_;
  size = (size_ < 0 ? INT_MAX : size_);
  delay = delay_;
  
  // Initialize lock
  mutex_threads_to_run = new pthread_mutex_t;
  pthread_mutex_init(mutex_threads_to_run, NULL);
}

ThreadPool::~ThreadPool() {
}

int ThreadPool::GetNewThreadID() {
  next_thread_id = (next_thread_id % INT_MAX) + 1;
  return next_thread_id;
}

void ThreadPool::Run() {
  debug("Threadpool %d: Running.\n", type);
  while (true) {
    debug("Threadpool %d: Cleaning out threads marked as done.\n", type);
    // Clean out the finished threads.
    for (map<int, int>::iterator it = threads_status.begin(); 
         it != threads_status.end();
         it++) {
      if (it->second == kThreadStatusMarkAsDone) {
        // Thread is done, delete it from the list of running_threads.
        Thread* finished_thread = running_threads[it->first];
        delete finished_thread;
        // Also erase it from the map of running threads.
        running_threads.erase(it->first);
        SetStatus(it->first, kThreadStatusDone);
        num_running_threads--;
      }
    }

    debug("Threadpool %d: Attempt to run new threads.\n", type);
    // Next, see if we can run some new threads.
    while (num_running_threads < size && !threads_to_run.empty()) {
      // Pop the next thread to run. FIFO style.
      pthread_mutex_lock(mutex_threads_to_run);
      Thread* next_thread = threads_to_run.front().second;
      int next_thread_id = threads_to_run.front().first;
      threads_to_run.pop();
      pthread_mutex_unlock(mutex_threads_to_run);

      // Update the maps and statuses.
      running_threads[next_thread_id] = next_thread;
      SetStatus(next_thread_id, kThreadStatusRunning);

      // Run the thread!
      debug("Threadpool %d: Starting thread %d (type %d).\n",
            type, next_thread_id, next_thread->GetType());
      debug("Threadpool %d: Size of ToDo list %d.\n", type,
            static_cast<int>(threads_to_run.size()));
      next_thread->Start(next_thread_id);
      num_running_threads++;
    }

    sleep(delay);
    sched_yield();
  }
}

int ThreadPool::AddThread(Thread *thread) {
  int id = GetNewThreadID();
  
  pthread_mutex_lock(mutex_threads_to_run);
  threads_to_run.push(pair<int, Thread*>(id, thread));
  pthread_mutex_unlock(mutex_threads_to_run);
  threads_status[id] = kThreadStatusQueued;
  debug("Threadpool %d: Add thread %d of type %d.\n", type, id,
        thread->GetType());
  return id;
}

void ThreadPool::SetStatus(int id, int status) {
  // Only allow threads that are already in the status map to be updated, since
  // these are threads that are being run or have already been run.
  if (threads_status.find(id) != threads_status.end()) {
    threads_status[id] = status;
    debug("Threadpool %d: Thread %d changed status to %d.\n", type, id, status);
  }
}

int ThreadPool::GetStatus(int id) {
  if (threads_status.find(id) == threads_status.end()) {
    return kThreadStatusUndefined;
  } else {
    return threads_status[id];
  }
}

bool ThreadPool::HasJobsRunning() {
  debug("Threadpool %d: Running: %d, enqueued: %d.\n", type,
        num_running_threads, static_cast<int>(threads_to_run.size()));
  return (num_running_threads || !threads_to_run.empty());
}
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=threadpool.h]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.
//
// Defines the ThreadPool class.
//
// Remember: compile using -pthread when using threads.

#ifndef THREADPOOL_H_
#define THREADPOOL_H_

#include <pthread.h>
#include <stdlib.h>
#include <time.h>
#include <unistd.h>

#include <map>
#include <queue>
#include <string>
#include <vector>

#include "thread.h"

using namespace std;

// Default thread pool size, i.e. max number of threads allowed to run
// concurrently.
const int kDefaultThreadPoolSize = 50;
// Default thread pool delay, i.e. number of seconds to wait in between checking
// to see if new threads can be executed.
const int kDefaultThreadPoolDelay = 1;

// A thread is not defined. 
const int kThreadStatusUndefined = 0;
// A thread has been added to a threadpool, but has not been run.
const int kThreadStatusQueued = 1;
// A thread is currently running.
const int kThreadStatusRunning = 2;
// A thread has finished running.
const int kThreadStatusDone = 3;
// A thread has just finished executing and is informing the threadpool before
// anything else.
const int kThreadStatusMarkAsDone = 4;

// Class representing thread pools. To prevent an excessive number of threads
// from being run concurrently, you shouldn't call Thread.Start() directly,
// instead, add it to a thread pool. The thread pool will ensure the number of
// threads being run (for each pool) is no more than a specified size. 
class ThreadPool {
  pthread_t thread;
  int type;
  pthread_mutex_t *mutex_threads_to_run;

  // Run the thread.
  static void *ThreadFunction(void *t) {
    ((ThreadPool*)t)->Run();
    return (void *)pthread_self();
  }

  // A queue containing a list of <id, Threads> that are not yet run.
  queue< pair<int, Thread*> > threads_to_run;
  // Maps thread ID's to threads which are already being run.
  map<int, Thread*> running_threads;
  // Maps thread ID's to their status.
  map<int, int> threads_status;  

  // Maximum number of threads to be allowed running concurrently in this pool.
  int size;
  // Number of seconds to wait in between updating the running threads.
  int delay;
  // The number of running threads at any given time.
  int num_running_threads;
  // The next thread ID to return. Basically a counter. Resets to zero when the
  // maximum size of int is reached.
  int next_thread_id;
  
  // Returns a new thread ID. Basically a counter.
  int GetNewThreadID();

 public:
  // Initializes all the class variables with default parameters.
  ThreadPool(int type_=kNullThreadType,
             int size_=kDefaultThreadPoolSize,
             int delay_=kDefaultThreadPoolDelay);

  // Empty destructor.
  ~ThreadPool();

  // Runs the thread pool. Will only stop once stop_when_done is set to true and
  // all the threads in the todo list are done.
  void Run();

  // Add a thread to the to do list.
  int AddThread(Thread *thread);

  // Set a status of a running or complete job. Will do nothing if another ID is
  // set.
  void SetStatus(int id, int status);

  // Returns the status of the thread with id.
  int GetStatus(int id);

  // Returns whether or not this threadpool has threads running or queued to
  // run.
  bool HasJobsRunning();

  // Start threadpool execution.
  int Start() {
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

    int tid = pthread_create(&thread, NULL, ThreadPool::ThreadFunction,
                             (void*)this);

    pthread_attr_destroy(&attr);
    return tid;
  }

  // Wait until Run() finishes doing its thing.
  int Wait() {
    return pthread_join(thread, NULL);
  }
};

#endif  // THREADPOOL_H_
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=wait.cc]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.

#include "wait.h"

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <unistd.h>

#include "common.h"
#include "threadpool.h"

using namespace std;

extern ThreadPool *job_thread_pool;
extern pthread_mutex_t *mutex_job_thread_pool;

void Wait(int job_id) {
  while (true) {
    pthread_mutex_lock(mutex_job_thread_pool);
    int status = job_thread_pool->GetStatus(job_id);
    pthread_mutex_unlock(mutex_job_thread_pool);

    if (status == kThreadStatusUndefined) {
      // TODO: Decide what to do if the user specifies a job thread ID which is
      // invalid. For now, it will exit.
      debug("Job thread ID %d is undefined.\n", job_id);
      break;
    } else if (status == kThreadStatusDone) {
      break;
    }

    sleep(kDefaultWaitDelay);
  }
}

void WaitAll() {
  while (true) {
    pthread_mutex_lock(mutex_job_thread_pool);
    bool has_jobs_running = job_thread_pool->HasJobsRunning();
    pthread_mutex_unlock(mutex_job_thread_pool);

    if (has_jobs_running) {
      sleep(kDefaultWaitDelay);
    } else {
      break;
    }
  }
}
\end{lstlisting}

\lstset{language=C++}
\begin{lstlisting}[caption=wait.h]
// Copyright 2009 Pigasus.
// GNU GPL v2.
// Author: John Zhang and Jordan Rupprecht.
//
// Implements the wait keyword for Pigasus.

#ifndef WAIT_H_
#define WAIT_H_

// Seconds to wait before checking again.
const int kDefaultWaitDelay = 1;

// Block the program while a specified job is still running.
void Wait(int job_id);

// Block the program while any job is still running.
void WaitAll();

#endif  // WAIT_H_
\end{lstlisting}

